using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Me.Prettyprint.Cassandra.Connection.Client;
using Me.Prettyprint.Cassandra.Connection.Factory;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Exceptions;
using Me.Prettyprint.Hector.Api.Factory;
using Org.Apache.Cassandra.Thrift;
using Me.Prettyprint.Cassandra.Utils;

namespace Me.Prettyprint.Cassandra.Connection
{
    /// <summary>
    /// Class is responsible for retrying downed hosts. When a host comes
    /// back up, the retry service will see it and re-add it.
    /// </summary>
	public class CassandraHostRetryService : BackgroundCassandraHostService
	{
		private static Logger logger = new Logger(typeof(CassandraHostRetryService));
		
		public const int DEF_QUEUE_SIZE = 100;
		public const int DEF_RETRY_DELAY = 5;
		
		private HClientFactory clientFactory;
		private List<CassandraHost> downedHostList;
		private ConnectionManagerListenersHandler listenerHandler;
		private bool autoDiscoverHosts;
        private CancellationTokenSource tokenSource = new CancellationTokenSource();
		
		public CassandraHostRetryService(HConnectionManager connectionManager, HClientFactory clientFactory, 
            CassandraHostConfigurator cassandraHostConfigurator, 
            ConnectionManagerListenersHandler listenerHandler) : base(connectionManager, cassandraHostConfigurator)
		{
			this.clientFactory = clientFactory;
			
			this.listenerHandler = listenerHandler;
			this.retryDelayInSeconds = cassandraHostConfigurator.RetryDownedHostsDelayInSeconds;
			this.autoDiscoverHosts = cassandraHostConfigurator.AutoDiscoverHosts;
            downedHostList = new List<CassandraHost>();

            var token = tokenSource.Token;
            backgroundTask = Task.Factory.StartNew(() => StartBackgroundProcess(token));
			string message = string.Format("Downed Host Retry service started with queue size {0} and retry delay {1}s", 
                cassandraHostConfigurator.RetryDownedHostsQueueSize, retryDelayInSeconds);
			logger.Info(message);
		}
        public void StartBackgroundProcess(CancellationToken token)
        {
            if (token.IsCancellationRequested)
            {
                token.ThrowIfCancellationRequested();
            }
            while (!token.IsCancellationRequested)
            {
                Thread.Sleep(retryDelayInSeconds * 1000);
                RetryDownedHosts();
            }
            if (token.IsCancellationRequested)
            {
                token.ThrowIfCancellationRequested();
            }
        }
        internal override void Shutdown()
		{
		}
        public List<CassandraHost> GetDownedHosts()
        {
            return this.downedHostList;
        }
		public void Add(CassandraHost cassandraHost)
		{
            if (!downedHostList.Contains(cassandraHost))
            {
                downedHostList.Add(cassandraHost);
                logger.Warning("Host " + cassandraHost.Ip + " is down. Adding to retry service.");
                //schedule a check of this host immediately,
                try
                {
                    CancellationTokenSource checkToken = new CancellationTokenSource();
                    var cToken = checkToken.Token;
                    Task checkTask = Task.Factory.StartNew(() => CheckHostAndAdd(cToken, cassandraHost));
                }
                catch (Exception e)
                {
                    logger.Error(e.Message);
                }
            }
		}
        public void CheckHostAndAdd(CancellationToken token, CassandraHost cassandraHost)
        {
            if (token.IsCancellationRequested)
            {
                token.ThrowIfCancellationRequested();
            }
            if (downedHostList.Contains(cassandraHost) && VerifyConnection(cassandraHost))
            {
                if (connectionManager.AddCassandraHost(cassandraHost))
                {
                    listenerHandler.FireOnHostRestored(cassandraHost);
                    downedHostList.Remove(cassandraHost);
                }
                return;
            }
        }
		public bool Remove(CassandraHost cassandraHost)
		{
			return downedHostList.Remove(cassandraHost);
		}
		public bool Contains(CassandraHost cassandraHost)
		{
			return downedHostList.Contains(cassandraHost);
		}
        /// <summary>
        /// Note: this doesn't really apply to .NET Task API
        /// </summary>
		public override void ApplyRetryDelay()
		{
		}
		public virtual void  flushQueue()
		{
			downedHostList.Clear();
			logger.Info("Downed Host retry queue flushed.");
		}
		public void RetryDownedHosts()
        {
            try
            {
                bool checkRing = connectionManager.GetHosts().Count > 0 && autoDiscoverHosts ? true : false;
                List<CassandraHost> ringInfo = null;
                if (checkRing)
                {
                    ringInfo = BuildRingInfo();
                    if (ringInfo != null && ringInfo.Count == 0)
                    {
                        ringInfo = null;
                    }
                }
                IEnumerator<CassandraHost> enumerator = downedHostList.GetEnumerator();
                while (enumerator.MoveNext())
                {
                    CassandraHost cassandraHost = enumerator.Current;
                    if (cassandraHost == null)
                    {
                        continue;
                    }
                    if (connectionManager.GetHosts().Count == 0)
                    {
                        listenerHandler.FireOnAllHostsDown();
                    }
                    if (checkRing && ringInfo != null && !ringInfo.Contains(cassandraHost))
                    {
                        downedHostList.Remove(cassandraHost);
                        logger.Info("removed " + cassandraHost.Ip + " from downedHostList.");
                        continue;
                    }
                    bool reconnected = VerifyConnection(cassandraHost);
                    if (reconnected)
                    {
                        connectionManager.AddCassandraHost(cassandraHost);
                        if (connectionManager.GetHosts().Contains(cassandraHost))
                        {
                            listenerHandler.FireOnHostRestored(cassandraHost);
                            downedHostList.Remove(cassandraHost);
                        }
                    }
                }
            }
            catch (Exception e)
            {
                logger.Error(e.StackTrace);
            }
        }
        private List<CassandraHost> BuildRingInfo()
        {
            ThriftCluster cluster = (ThriftCluster) HFactory.GetCluster(connectionManager.ClusterName);
            // ThriftCluster is not exclusively created & cached by HFactory. E.g. Some users instantiate directly via Spring.
            if(cluster != null)
            {
                List<CassandraHost> ringInfo = new List<CassandraHost>();
                foreach (KeyspaceDefinition keyspaceDefinition in cluster.DescribeKeyspaces())
                {
                    if (!keyspaceDefinition.Name.Equals(Me.Prettyprint.Hector.Api.KeyspaceFields.KEYSPACE_SYSTEM))
                    {
                        List<TokenRange> tokenRanges = cluster.DescribeRing(keyspaceDefinition.Name);
                        foreach (TokenRange tokenRange in tokenRanges)
                        {
                            foreach (String host in tokenRange.Rpc_endpoints)
                            {
                                CassandraHost aHost = new CassandraHost(host, cassandraHostConfigurator.Port);
                                if (!ringInfo.Contains(aHost) )
                                {
                                    ringInfo.Add(aHost);
                                }
                            }
                        }
                        break;
                    }
                }
                return ringInfo;
            } else {
                return null;
            }
        }
		private bool VerifyConnection(CassandraHost cassandraHost)
		{
			if (cassandraHost == null)
			{
				return false;
			}
			bool found = false;
			HClient client = clientFactory.createClient(cassandraHost);
            if (client != null)
            {
                try
                {
                    client.Open();
                    found = client.GetCassandra().describe_cluster_name() != null;
                    client.Close();
                }
                catch (HectorTransportException he)
                {
                    logger.Warning(string.Format("Downed {0} host still appears to be down: {1}", cassandraHost, he.Message));
                }
                catch (System.Exception t)
                {
                    logger.Error(string.Format("Downed Host retry failed attempt to verify CassandraHost {0}", t));
                }
            }
			return found;
		}
	}
}